package network

import (
	"errors"
	"github.com/gorilla/websocket"
	"leaf/core/log"
	"leaf/core/msg"
	"leaf/core/tool"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type WsConn struct {
	mutex     sync.Mutex
	addr      string
	conn      *websocket.Conn
	writeChan chan []byte
	closeFlag bool
	rflag     int32
	cflag     int32
	dflag     int32
	msgParser *MsgParser
	rwg       sync.WaitGroup
	cwg       sync.WaitGroup
	connType  int32
	server    *WsServer
	agent     Agent

	dialer           websocket.Dialer
	HandshakeTimeout time.Duration
}

func newAcceptWsConn(conn *websocket.Conn, msgParser *MsgParser, server *WsServer) *WsConn {
	wsConn := new(WsConn)
	wsConn.conn = conn
	if conn == nil {
		log.Error("new accept WsConn err: conn is nil")
		return nil
	}
	wsConn.writeChan = make(chan []byte, 1025)
	wsConn.msgParser = msgParser
	if wsConn.msgParser == nil {
		wsConn.msgParser = DefaultMsgParser
	}
	wsConn.connType = int32(CONN_TYPE_ACCEPT)
	wsConn.server = server
	if wsConn.HandshakeTimeout <= 0 {
		wsConn.HandshakeTimeout = 10 * time.Second
	}
	if wsConn.server == nil {
		log.Error("new accept WsConn err: server is nil")
		return nil
	}
	return wsConn
}

func NewWsConn(addr string, msgParser *MsgParser, agent Agent) *WsConn {
	wsConn := new(WsConn)
	wsConn.addr = addr
	wsConn.writeChan = make(chan []byte, 1025)
	wsConn.msgParser = msgParser
	if wsConn.msgParser == nil {
		wsConn.msgParser = DefaultMsgParser
	}
	wsConn.connType = int32(CONN_TYPE_CONNECT)
	wsConn.agent = agent
	if wsConn.HandshakeTimeout <= 0 {
		wsConn.HandshakeTimeout = 10 * time.Second
	}
	wsConn.dialer = websocket.Dialer{
		HandshakeTimeout: wsConn.HandshakeTimeout,
	}
	if wsConn.agent == nil {
		log.Error("new WsConn err: agent is nil")
		return nil
	}
	if wsConn.server == nil {
		log.Error("new WsConn err: server is nil")
		return nil
	}
	return wsConn
}

func (wsConn *WsConn) GetAgent() Agent {
	return wsConn.agent
}

func (wsConn *WsConn) dial() *websocket.Conn {
	for {
		if wsConn.cflag != 1 {
			log.Release("wsConn dial flag false")
			return nil
		}

		conn, _, err := wsConn.dialer.Dial(wsConn.addr, nil)
		if err == nil {
			if wsConn.cflag != 1 {
				log.Release("wsConn dial flag false")
				conn.Close()
				return nil
			}
			return conn
		}

		log.Release("connect to %v error: %v", wsConn.addr, err)
		time.Sleep(100 * time.Millisecond)
		continue
	}
}

func (wsConn *WsConn) run() bool {
	if wsConn.conn == nil {
		log.Error("wsConn run err:", "conn is nil")
		return false
	}
	if wsConn.agent == nil {
		log.Error("wsConn run err:", "agent is nil")
		return false
	}
	if !atomic.CompareAndSwapInt32(&wsConn.rflag, 0, 1) {
		log.Error("wsConn run err:", "rflag not zero")
		return false
	}
	wsConn.rwg.Add(2)
	tool.Go(func() {
		defer func() {
			wsConn.rwg.Add(-1)
			wsConn.runOver() //让读写线程退出
		}()

		for tool.IsRuning() && wsConn.rflag == 1 {
			data, err := wsConn.ReadMsg()
			if err != nil {
				log.Error("wsConn read err: %v", err)
				return
			}
			head, bytes, err := msg.Unmarshal(data)
			if err != nil {
				log.Error("wsConn read err: %v", err)
			}
			if wsConn.agent != nil {
				wsConn.agent.Process(head, bytes)
			}
		}
	})

	tool.Go(func() {
		defer func() {
			wsConn.rwg.Add(-1)
			wsConn.runOver() //让读写线程退出
		}()

		for tool.IsRuning() {
			var data []byte
			select {
			case <-tool.GoSigChan():
			case data = <-wsConn.writeChan:
				if data != nil {
					err := wsConn.conn.WriteMessage(websocket.BinaryMessage, data)
					if err != nil {
						log.Error("wsConn write err: %v", err)
						return
					}
				}
			}
			if data == nil {
				return
			}
		}
	})

	return true
}

func (wsConn *WsConn) connect() bool {
	if wsConn.agent == nil {
		log.Error("wsConn run err:", "agent is nil")
		return false
	}
	if !atomic.CompareAndSwapInt32(&wsConn.cflag, 0, 1) {
		log.Error("wsConn run err:", "cflag not zero")
		return false
	}
	wsConn.cwg.Add(1)
	tool.Go(func() {
		defer func() {
			wsConn.cwg.Add(-1)
		}()

		if tool.IsRuning() && wsConn.cflag == 1 {
			conn := wsConn.dial()
			if conn == nil {
				return
			}
			conn.SetReadLimit(int64(Default_MaxMsgLen))
			wsConn.conn = conn
			wsConn.run()

			//等待读写线程都退出
			wsConn.rwg.Wait()
			wsConn.conn.Close()
			wsConn.conn = nil

			time.Sleep(CONNECT_INTERVAL)
		}
	})

	return true
}

func (wsConn *WsConn) Start() bool {
	if wsConn.connType == CONN_TYPE_ACCEPT {
		return wsConn.run()
	} else if wsConn.connType == CONN_TYPE_CONNECT {
		return wsConn.connect()
	} else {
		log.Error("ws conn start err:%v connType:%v", "conn type wrong", wsConn.connType)
		return false
	}
}

func (wsConn *WsConn) runOver() {
	if atomic.CompareAndSwapInt32(&wsConn.rflag, 1, 0) {
		wsConn.conn.SetReadDeadline(time.Now())
		wsConn.conn.SetWriteDeadline(time.Now())
		wsConn.writeChan <- nil
		if wsConn.server != nil {
			wsConn.server.onCloseConn(wsConn)
		}
	}
}

func (wsConn *WsConn) Close() {
	if atomic.CompareAndSwapInt32(&wsConn.rflag, 1, 0) {
		wsConn.conn.SetReadDeadline(time.Now())
		wsConn.conn.SetWriteDeadline(time.Now())
		wsConn.writeChan <- nil

		if wsConn.server != nil {
			wsConn.server.onCloseConn(wsConn)
		}
	}

	if atomic.CompareAndSwapInt32(&wsConn.cflag, 1, 0) {

	}
}

func (wsConn *WsConn) Destroy() {
	wsConn.rwg.Wait()
	wsConn.cwg.Wait()
	if atomic.CompareAndSwapInt32(&wsConn.dflag, 0, 1) {
		close(wsConn.writeChan)
		if wsConn.conn != nil {
			wsConn.conn.Close()
			wsConn.conn = nil
		}
	}
}

func (wsConn *WsConn) doWrite(b []byte) {
	if len(wsConn.writeChan) == cap(wsConn.writeChan) {
		log.Debug("close conn: channel full")
		wsConn.Close()
		return
	}

	wsConn.writeChan <- b
}

func (wsConn *WsConn) LocalAddr() net.Addr {
	return wsConn.conn.LocalAddr()
}

func (wsConn *WsConn) RemoteAddr() net.Addr {
	return wsConn.conn.RemoteAddr()
}

// goroutine not safe
func (wsConn *WsConn) ReadMsg() ([]byte, error) {
	_, b, err := wsConn.conn.ReadMessage()
	return b, err
}

// args must not be modified by the others goroutines
func (wsConn *WsConn) WriteMsg(args ...[]byte) error {
	wsConn.mutex.Lock()
	defer wsConn.mutex.Unlock()
	if wsConn.closeFlag {
		return nil
	}

	// get len
	var msgLen uint32
	for i := 0; i < len(args); i++ {
		msgLen += uint32(len(args[i]))
	}

	// check len
	if msgLen > Default_MaxMsgLen {
		return errors.New("message too long")
	} else if msgLen < 1 {
		return errors.New("message too short")
	}

	// don't copy
	if len(args) == 1 {
		wsConn.doWrite(args[0])
		return nil
	}

	// merge the args
	msg := make([]byte, msgLen)
	l := 0
	for i := 0; i < len(args); i++ {
		copy(msg[l:], args[i])
		l += len(args[i])
	}

	wsConn.doWrite(msg)

	return nil
}
